package test.top2.com.atguigu.app.fun;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.JDBCUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import test.top2.com.atguigu.bean.TableProcess;

import java.sql.Connection;
import java.sql.DriverManager;
import java.util.*;

/**
 * ClassName: DwdTableProcessFuntion
 * Package: test.top2.com.atguigu.app.fun
 * Description:
 *
 * @Author ChenJun(有志男青年)
 * @Create 2023/5/4 21:06
 * @Version 1.0
 */
public class DwdTableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    private HashMap<String, TableProcess> tableProcessHashMap;

    public DwdTableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    //预加载
    @Override
    public void open(Configuration parameters) throws Exception {

        tableProcessHashMap = new HashMap<>();

        //使用jdbc方式读取配置信息表，将数据存入hashmap
        Connection connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/gmall_config?" + "user=root&password=000000&useUnicode=true&" + "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");

        List<TableProcess> tableProcesses = JDBCUtil.queryList(connection, "select * from table_process where sink_type='dwd'", TableProcess.class, true);

        for (TableProcess tableProcess : tableProcesses) {
            String key = tableProcess.getSourceTable() + "-" + tableProcess.getSourceType();
            tableProcessHashMap.put(key, tableProcess);
        }

        connection.close();
    }

    //广播流
    //value:{"before":null,"after":{"source_table":"base_category3","sink_table":"dim_base_category3","sink_columns":"id,name,category2_id","sink_pk":"id","sink_extend":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1669162876406,"snapshot":"false","db":"gmall-220623-config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1669162876406,"transaction":null}
    @Override
    public void processBroadcastElement(String value, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {

        //0.获取状态信息
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        //1.如果为删除操作,则删除状态以及Map中的配置信息
        JSONObject jsonObject = JSONObject.parseObject(value);
        if ("d".equals(jsonObject.getString("op"))){
            TableProcess tableProcess = JSONObject.parseObject(jsonObject.getString("before"), TableProcess.class);
            String key = tableProcess.getSourceTable()+"-"+tableProcess.getSourceType();

            broadcastState.remove(key);
            tableProcessHashMap.remove(key);

        }else{
            //2.转换为tableProcess对象
            TableProcess tableProcess = JSONObject.parseObject(jsonObject.getString("after"), TableProcess.class);

            if ("dwd".equals(tableProcess.getSinkType())){
                String key = tableProcess.getSourceTable()+"-"+tableProcess.getSourceType();

                System.out.println("processBroadcastElement--Key:" + key);

                //3.写入状态
                broadcastState.put(key,tableProcess);
            }
        }
    }

    //主流
    //Value:{"database":"gmall-221109-flink","table":"comment_info","type":"insert","ts":1669162958,"xid":1111,"xoffset":13941,"data":{"id":1595211185799847960,"user_id":119,"nick_name":null,"head_img":null,"sku_id":31,"spu_id":10,"order_id":987,"appraise":"1204","comment_txt":"评论内容：48384811984748167197482849234338563286217912223261","create_time":"2022-08-02 08:22:38","operate_time":null}}
    @Override
    public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //从广播状态中提取对应的配置信息
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        String table = value.getString("table");
        String key = table +"-"+value.getString("type");
        System.out.println("processElement--Key:" + key);

        TableProcess tableProcess = broadcastState.get(key);
        TableProcess tableProcessMap = tableProcessHashMap.get(key);

        //2.1 过滤数据(行过滤)
        if (tableProcess != null|| tableProcessMap != null){

            if (tableProcess == null){
                tableProcess=tableProcessMap;
            }

            //2.2 过滤数据(列过滤)
            filterColumns(value.getJSONObject("data"),tableProcess.getSinkColumns());

            //3.补充SinkTable字段将数据写出
            value.put("sink_table",tableProcess);
            out.collect(value);

        }else {
            System.out.println("该表：" + table + "没有对应的配置信息！");
        }

    }

    private void filterColumns(JSONObject data, String sinkColumns) {
        String[] columnArr = sinkColumns.split(",");
        List<String> columnsList = Arrays.asList(columnArr);

        Set<Map.Entry<String, Object>> entries = data.entrySet();
        entries.removeIf(entry -> !columnsList.contains(entry.getKey()));
    }


}
